package com.atguigu.mybatisplus.web;

import com.atguigu.mybatisplus.entity.User;
import com.atguigu.mybatisplus.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.IntStream;

/**
 * @author zhuzi
 * 2022/8/14 15:26
 * 在类上添加 Transactiona 可以解决同类中没有添加事务注解的方法调用添加了事务注解的方法 这总事务的控制
 */
@Transactional(propagation = Propagation.SUPPORTS, readOnly = true)
@Slf4j
@RestController
@RequestMapping(path = "txThread")
public class TxThreadController {
    @Autowired
    UserService userService;

    /**
     * 装配mybatis-springboot 的事务管理器,
     * 也可以替代成springboot-jdbc开启,
     * 也可以替代成sqlsessionfactorybean开启
     */
    @Resource
    private DataSourceTransactionManager txManager;

    /**
     * CompletableFuture 多线程下操作数据库的事务问题
     */
    @GetMapping
    public void threadTx() {
        //定义开启的线程数
        AtomicInteger totalThreadCount = new AtomicInteger(2);
        // 事务定义
        DefaultTransactionDefinition def = new DefaultTransactionDefinition();

        //每次都开启一个新事物
        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

        //所有开启的事务的线程中，是否存在异常
        AtomicBoolean isException = new AtomicBoolean(Boolean.FALSE);

        //模拟多线程访问
        ArrayList<CompletableFuture<Void>> list = new ArrayList<>();
        //定义没有结束线程集合
        List<Thread> unFinishedThread = Collections.synchronizedList(new ArrayList<>());

        IntStream.range(0, totalThreadCount.get()).forEach(value ->
                list.add(CompletableFuture.runAsync(() -> {
                    //获得事务状态
                    TransactionStatus status = txManager.getTransaction(def);
                    try {
                        userService.save(new User().setAge(27).setName(RandomStringUtils.randomAlphabetic(5)).setId((long) (8 + value)));
                        if (value == 1) {
                            throw new RuntimeException("异常发生");
                        }
                        //添加没执行结束的线程
                        unFinishedThread.add(Thread.currentThread());
                        // 每个线程都在悬停前开启唤醒检查
                        log.info("即将开启调用:{},{}", unFinishedThread.size(), totalThreadCount.get());
                        this.notifyAllThread(unFinishedThread, totalThreadCount, false);
                        log.info("即将悬停线程{}", Thread.currentThread());
                        LockSupport.park();
                        // 根据结果, 令设定各个线程的行为
                        if (isException.get()) {
                            txManager.rollback(status);
                        } else {
                            txManager.commit(status);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        // 本线程回滚
                        txManager.rollback(status);
                        // 设定线程中存在异常信息
                        isException.set(Boolean.TRUE);
                        this.notifyAllThread(unFinishedThread, totalThreadCount, true);
                    }
                }))

        );
        CompletableFuture<Void> result = CompletableFuture.allOf(list.toArray(new CompletableFuture[0]));
        try {
            // 调用Future的阻塞接口, 等待全部future实例异步执行结束
            result.get();
        } catch (Exception e) {
            throw new RuntimeException("子线程中抛出异常");
        }

    }

    /**
     * 唤醒全部悬停的线程
     *
     * @param unFinishedThread 手动悬停的线程
     * @param totalThreadCount 全部开启的线程数
     * @param isForce          是否强行操作集合中全部线程
     */
    private void notifyAllThread(List<Thread> unFinishedThread, AtomicInteger totalThreadCount, boolean isForce) {
        // 当手动悬停的线程与开启的线程数相同时, 证明全部线程都已经开启并手动悬停, 故而可以唤醒全部手动悬停的线程, 并处理提交与回滚逻辑
        if (isForce || unFinishedThread.size() == totalThreadCount.get()) {
            for (Thread thread : unFinishedThread) {
                log.info("唤醒线程{}", thread);
                LockSupport.unpark(thread);
            }
        }
    }
}
